package com.douxg.learning.springboot.es;

import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.xpack.sql.jdbc.EsDataSource;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

/**
 * @Description
 * @Auther douxg
 * @Date 2022/8/23 19:08
 */
@Slf4j
@Configuration
@ConfigurationProperties("es")
@Data
public class EsConfig {
    private String address = "127.0.0.1:9200";
    /**
     * 连接超时时间
     */
    private int connectTimeOut = 1000;
    /**
     * 连接超时时间
     */
    private int socketTimeOut = 30000;
    /**
     * 获取连接的超时时间
     */
    private int connectionRequestTimeOut = 30000;
    /**
     * 最大连接数
     */
    private int maxConnectNum = 100;
    /**
     * 最大路由连接数
     */
    private int maxConnectPerRoute = 100;
    private int keepAliveTime = 30000;

    @Bean
    public ElasticsearchAsyncClient createElasticsearchAsyncClient(RestClientTransport restClientTransport) {
        return new ElasticsearchAsyncClient(restClientTransport);
    }

    @Bean
    public ElasticsearchClient createElasticsearchClient(RestClientTransport restClientTransport) {
        return new ElasticsearchClient(restClientTransport);
    }

    /*@Bean
    public EsDataSource createEsDataSource() {
        EsDataSource dataSource = new EsDataSource();
        String address = "jdbc:es://http://" + this.address;
        dataSource.setUrl(address);
        Properties connectionProperties = new Properties();
        dataSource.setProperties(connectionProperties);
        return dataSource;
    }*/

    @Bean
    public RestClient createRestClient() {
        List<HttpHost> httpAddresses = getHosts(Arrays.asList(address.split(",")));
        RestClientBuilder restClientBuilder = RestClient.builder(httpAddresses.toArray(new HttpHost[httpAddresses.size()]));
        restClientBuilder.setRequestConfigCallback(
                b -> b.setConnectTimeout(connectTimeOut)
                        .setConnectionRequestTimeout(connectionRequestTimeOut)
                        .setSocketTimeout(socketTimeOut));
        restClientBuilder.setHttpClientConfigCallback(
                b -> b.setKeepAliveStrategy((response, context) -> keepAliveTime)
                        .setMaxConnPerRoute(maxConnectPerRoute));
        return restClientBuilder.build();
    }

    @Bean
    public RestClientTransport createRestClientTransport(RestClient restClient) {
        return new RestClientTransport(restClient, new JacksonJsonpMapper());
    }

    private static List<HttpHost> getHosts(List<String> hosts) {
        return hosts.stream()
                .map(host -> validateAndParseHostsString(host))
                .collect(Collectors.toList());
    }

    private static HttpHost validateAndParseHostsString(String host) {
        HttpHost httpHost = HttpHost.create(host);
        if (httpHost.getPort() < 0) {
            throw new RuntimeException(String.format("Could not parse host '%s'. ", host));
        }

        if (httpHost.getSchemeName() == null) {
            throw new RuntimeException(String.format("Could not parse host '%s'.", host));
        }
        return httpHost;
    }
}
